#include <assert.h>
#include <google/protobuf/message.h>
#include <google/protobuf/text_format.h>

#include "echo.pb.h"
#include "zrpc/base/buffer.h"
#include "zrpc/base/logger.h"
#include "zrpc/net/event_loop.h"
#include "zrpc/net/tcp_client.h"
#include "zrpc/net/tcp_connection.h"
#include "zrpc/rpc/buffer_stream.h"

using namespace zrpc;

class EchoClient {
 public:
  EchoClient(std::string ip, int port)
      : client_(&loop_, ip, port, nullptr), count_(0) {}

  const static uint32_t kMsgHeadSize = static_cast<uint32_t>(sizeof(Message));

  const char* GetMessageBuf(const zrpc::Message& msg) {
    if (msg.size < kMsgHeadSize) {
      return nullptr;
    }

    return (reinterpret_cast<const char*>(&msg) + kMsgHeadSize);
  }

  void Start() {
    std::string str;
    for (int i = 0; i < 30000; ++i) {
      message_.push_back(static_cast<char>(i % 128));
    }

    client_.EnableRetry();
    // 设置连接回调注册监听
    client_.SetConnectionCallback(std::bind(&EchoClient::ConnectionCallback,
                                            this, std::placeholders::_1));
    // 设置消息回调注册监听
    client_.SetMessageCallback(std::bind(&EchoClient::MessageCallback, this,
                                         std::placeholders::_1,
                                         std::placeholders::_2));
    client_.Connect();
  }

  void ConnectionCallback(const std::shared_ptr<TcpConnection>& conn) {
    if (conn->Connected()) {
      assert(socket::SetKeepAlive(conn->GetSockfd(), 1));
      /*Buffer buf;
      echo::PingRequest request;
      request.set_id(message_);
      std::string str;
      assert(request.SerializeToString(&str));
      buf.Append(str);
      buf.PrependInt16(0);
      buf.PrependInt16(buf.ReadableBytes() + sizeof(int16_t));
      conn->Send(&buf);
      count_++;
	  */
      LOG_INFO << "连接服务器成功";
    } else {
      LOG_INFO << "断开服务器连接";
    }
  }

  void Loop() { loop_.Run(); }

  void MessageCallback(const std::shared_ptr<TcpConnection>& conn,
                       Buffer* buf) {
    while (buf->ReadableBytes() >= sizeof(uint32_t)) {
      int16_t data_bytes = buf->PeekInt16();
      if (buf->ReadableBytes() >= data_bytes) {
        buf->RetrieveInt32();
        //echo::PongRepsonse response;
        //assert(response.ParseFromArray(buf->Peek(), data_bytes - kMsgHeadSize));
        buf->Retrieve(data_bytes - kMsgHeadSize);

        if (count_++ < 500000) {
          Buffer buf;
          echo::PingRequest request;
          request.set_id(message_);
          std::string str;
          assert(request.SerializeToString(&str));
          buf.Append(str);
          buf.PrependInt16(0);
          buf.PrependInt16(buf.ReadableBytes() + sizeof(int16_t));
          conn->Send(&buf);
        } else {
          LOG_INFO << "cast time:"
                   << time(0) - std::any_cast<time_t>(conn->GetContext());
        }
      } else {
        break;
      }
    }
  }

  void TimerCallback() { LOG_INFO << "repeat timer callback"; }

 private:
  EventLoop loop_;
  TcpClient client_;
  std::string message_;
  uint32_t count_;
};

int main() {
  EchoClient client("10.23.7.160", 6379);
  client.Start();
  client.Loop();
}